package me.zhengjie.common.utils.batchsave.impl

import com.google.common.collect.Lists
import me.zhengjie.logging.slf4j.*
import me.zhengjie.common.utils.batchsave.vo.CallableResultVo
import me.zhengjie.common.utils.batchsave.callable.PartSaveCallable
import me.zhengjie.common.utils.batchsave.BatchSaveRepository
import jakarta.persistence.EntityManager
import jakarta.persistence.PersistenceContext
import org.springframework.stereotype.Repository
import org.springframework.transaction.annotation.Propagation
import org.springframework.transaction.annotation.Transactional
import java.io.Serializable
import java.util.concurrent.CountDownLatch
import java.util.concurrent.ExecutionException
import java.util.concurrent.Executors
import java.util.concurrent.ExecutorService
import java.util.concurrent.Future
import java.util.function.Consumer

//@NoRepositoryBean
@Repository
class BatchSaveRepositoryImpl<T, ID : Serializable?>  /*    public BatchSaveRepositoryImpl(JpaEntityInformation entityInformation, EntityManager entityManager) {
        super(entityInformation, entityManager);
        this.em = entityManager;
    }*/
    : BatchSaveRepository<T> {
    //每个线程分的数据量
    private val BATCH_SIZE = 1500

    //最大线程数
    private val MAX_THREAD = 12

    //private static EntityManager em = null;
    @PersistenceContext
    private val em: EntityManager? = null

    //@Async
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    override fun <S : T?> batchSave(entities: Iterable<S>?, isSave: Boolean?): List<S> {
        //返回list数组，需要带id
        val result: MutableList<S> = ArrayList()
        val lists: MutableList<S> = Lists.newArrayList()
        var listsTmp: List<S>
        entities!!.forEach(Consumer { e: S -> lists.add(e) })
        val listSize = lists.size

        //构造线程池 - 默认最大MaxThread个线程同时执行，每个线程执行数据量BATCH_SIZE
        val executorService: ExecutorService = Executors.newFixedThreadPool(MAX_THREAD)
        //需要循环测次数，最后一次大概率不满足一个BATCH_SIZE
        val loopCount = listSize / BATCH_SIZE + 1
        //倒计时门闩 - await() 让线程等待，用countDown()消初始化数量。当数量等于0时线程唤醒
        val cdl = CountDownLatch(loopCount) //使用计数器
        //创建FutureList,存储每一个线程返回的结果
        val futureSaveList: MutableList<Future<*>> = ArrayList()
        val futureUpdateList: MutableList<Future<*>> = ArrayList()

        //一共循环threadNum次
        for (i in 0 until loopCount) {
            if (i == loopCount - 1) {
                //走到头但不足一整次的部分
                logInfo(
                    "------------------------------------------------拆分数据最后一部分下标范围：start - {${i * BATCH_SIZE}}, end - {$listSize}."
                )
                listsTmp = lists.subList(i * BATCH_SIZE, listSize)
                val psc: PartSaveCallable<S> = PartSaveCallable(listsTmp, cdl, isSave!!)
                if (isSave!!) {
                    futureSaveList.add(executorService.submit<Any>(psc))
                } else {
                    futureUpdateList.add(executorService.submit<Any>(psc))
                }
            } else {
                logInfo(
                    "------------------------------------------------拆分数据下标范围：start - {${i * BATCH_SIZE}}, end - {${(i + 1) * BATCH_SIZE}}."
                )
                listsTmp = lists.subList(i * BATCH_SIZE, (i + 1) * BATCH_SIZE)
                val psc: PartSaveCallable<S> = PartSaveCallable(listsTmp, cdl, isSave!!)
                if (isSave!!) {
                    futureSaveList.add(executorService.submit<Any>(psc))
                } else {
                    futureUpdateList.add(executorService.submit<Any>(psc))
                }
            }
        }
        try {
            //确保线程执行完
            cdl.await()
            val tempFutureList: List<Future<*>> = if (isSave!!) futureSaveList else futureUpdateList
            for (future in tempFutureList) {
                //线程到这儿必定执行完了
                try {
                    val res = future.get()
                    if (res != null) {
                        val crv: CallableResultVo<S> = res as CallableResultVo<S>
                        result.addAll(crv.result!!)
                    }
                } catch (e: ExecutionException) {
                    e.printStackTrace()
                }
            }
        } catch (e: InterruptedException) {
            e.printStackTrace()
        } finally {
            //执行完关闭线程池
            executorService.shutdown()
        }
        return result
    }

    private fun <S> partBatchSave(entities: Iterable<S>): List<S> {
        val iterator = entities.iterator()
        var index = 0
        while (iterator.hasNext()) {
            em!!.persist(iterator.next())
            index++
            if (index % 500 == 0) {
                em.flush()
                em.clear()
            }
        }
        if (index % 500 != 0) {
            em!!.flush()
            em.clear()
        }
        val lists: MutableList<S> = Lists.newArrayList()
        entities.forEach(Consumer { e: S -> lists.add(e) })
        return lists
    }
}